package com.cnzha.mr.friend;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.*;

/**
 * @Author: zhaoxin
 * @email: zdkk@foxmail.com
 * @Date: 2020/4/6 18:34
 * A:B,C,D,F,E,O
 * B:A,C,E,K
 * C:F,A,D,I
 * D:A,E,F,L
 * E:B,C,D,M,L
 * F:A,B,C,D,E,O,M
 * G:A,C,D,E,F
 * H:A,C,D,E,O
 * I:A,O
 * J:B,O
 * K:A,C,D
 * L:D,E,F
 * M:E,F,G
 * O:A,H,I,J
 * 1:先输出每个人的双向好友集合
 */
public class FriendMapper extends Mapper<LongWritable, Text, Text, Text> {

    Map<String, List<String>> map = new HashMap();
    Text k = new Text();
    Text v = new Text();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        if (StringUtils.isEmpty(line)) {
            return;
        }
        String name = line.substring(0, 1);
        String[] split = line.substring(2).split(",");
        map.put(name, Arrays.asList(split));
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        ArrayList<HashSet> keyList = new ArrayList<>();
        for (String name : map.keySet()) {
            List<String> friendList = map.get(name);
            for (String frName : map.keySet()) {
                if (Objects.equals(name, frName)) {
                    continue;
                }
                k.set(name + "-" + frName);
                HashSet<Object> set = new HashSet<>(2);
                set.add(name);
                set.add(frName);
                List<String> frFriendList = map.get(frName);
                String tempText = "";
                for (String frNa : friendList) {
                    if (frFriendList.contains(frNa)) {
                        tempText += frNa + "\t";
                    }
                }
                if (StringUtils.isBlank(tempText)) {
                    continue;
                }
                if (keyList.contains(set)) {
                    continue;
                }
                keyList.add(set);
                v.set(tempText);
                context.write(k, v);
            }
        }
    }
}
